use std::io;
use std::io::{Read, Write};
use std::net::TcpStream;
use std::sync::{Arc, RwLock};
use std::thread::{JoinHandle, sleep, spawn};
use std::time::Duration;

pub enum StreamIO {
    ReadIO(ReadStream),
    WriteIO(WriteStream),
}

pub struct ReadStream {
    pub read_stream: Arc<RwLock<TcpStream>>,
    pub queue: Arc<RwLock<Vec<u8>>>,
}

impl ReadStream {
    pub fn spawn_read(&mut self) {
        read_socket_thread(self.read_stream.clone(), self.queue.clone());
    }
}
pub struct WriteStream {
    pub write_stream: Arc<RwLock<TcpStream>>,
    pub queue: Arc<RwLock<Vec<u8>>>,
}
impl WriteStream{
    pub fn spawn_write(&mut self){
        write_socket_stream(self.write_stream.clone(),self.queue.clone());
    }
}
impl Write for WriteStream {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        let mut guard = self.queue.write().unwrap();
        guard.extend_from_slice(buf);
        drop(guard);
        Ok(buf.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

impl Iterator for ReadStream {
    type Item = Vec<u8>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            let mut guard = self.queue.write().unwrap();
            if guard.is_empty() {
                drop(guard);
                sleep(Duration::from_millis(10));
                continue;
            };
            let item = guard.clone();
            guard.clear();
            drop(guard);
            return Some(item);
        }
    }
}
//线程长时间后台运行
pub fn read_socket_thread(
    mut stream: Arc<RwLock<TcpStream>>,
    output_queue: Arc<RwLock<Vec<u8>>>,
) -> JoinHandle<()> {
    spawn(move || {
        let mut buffer = [0u8; 1024];
        loop {
            let mut stream_guard = stream.write().unwrap();
            match stream_guard.read(&mut buffer) {
                Ok(0) => break,
                Ok(n) => {
                    let mut guard = output_queue.write().unwrap();
                    guard.extend_from_slice(&buffer[0..n]);
                    drop(guard);
                    drop(stream_guard);
                }
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    drop(stream_guard);
                    sleep(Duration::from_millis(10)); // 等待数据到达
                    continue;
                }
                Err(e) => panic!("Read error: {}", e),
            }
        }
    })
}

//线程长时间后台运行
pub fn write_socket_stream(
    mut stream: Arc<RwLock<TcpStream>>,
    input_queue: Arc<RwLock<Vec<u8>>>,
) -> JoinHandle<()> {
    spawn(move || {
        loop {
            let mut guard = input_queue.write().unwrap();
            let data = guard.clone();
            guard.clear();
            drop(guard);
            for chunk in data.chunks(10) {
                loop {
                    let mut stream_guard = stream.write().unwrap();
                    match stream_guard.write(chunk) {
                        Ok(_n) => break,
                        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                            drop(stream_guard);
                            sleep(Duration::from_millis(10)); // 等待缓冲区可用
                            continue;
                        }
                        Err(e) => panic!("Write error: {}", e),
                    }
                }
            }
        }
    })
}
